package com.atguigu.gmall.realtime.dwd.db.split.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcessDwd;
import com.atguigu.gmall.realtime.common.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.*;

/**
 * ClassName: BaseDbTableProcessFunction
 * Description:
 * 事实表动态分流处理逻辑
 * @Create 2024/4/22 16:07
 */
public class BaseDbTableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>> {

    Map<String, TableProcessDwd> configMap = new HashMap<>();

    MapStateDescriptor<String, TableProcessDwd> mapStateDescriptor;

    public BaseDbTableProcessFunction(MapStateDescriptor<String, TableProcessDwd> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        Connection mysqlConnection = JdbcUtil.getMysqlConnection();
        List<TableProcessDwd> tableProcessDwds = JdbcUtil.queryList(mysqlConnection, "select * from gmall1030_config.table_process_dwd", TableProcessDwd.class, true);

        for (TableProcessDwd tableProcessDwd : tableProcessDwds) {
            String sourceTable = tableProcessDwd.getSourceTable();
            String sourceType = tableProcessDwd.getSourceType();
            String key = getKey(sourceTable, sourceType);
            configMap.put(key,tableProcessDwd);
        }

        JdbcUtil.closeConnection(mysqlConnection);
    }

    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>>.ReadOnlyContext readOnlyContext, Collector<Tuple2<JSONObject, TableProcessDwd>> collector) throws Exception {
        ReadOnlyBroadcastState<String, TableProcessDwd> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);

        //获取业务表名
        String table = jsonObject.getString("table");
        //获取对业务数据的操作
        String type = jsonObject.getString("type");
        //拼接
        String key = getKey(table, type);
        //根据key到广播状态获取配置信息
        TableProcessDwd tableProcessDwd = null;

        if ((tableProcessDwd = broadcastState.get(key)) != null || (tableProcessDwd = configMap.get(key)) != null) {

            JSONObject dataJsonObj = jsonObject.getJSONObject("data");

            String sinkColumns = tableProcessDwd.getSinkColumns();

            deleteNotNeedColums(dataJsonObj,sinkColumns);

            //在向下游传递数据前，补充ts
            Long ts = jsonObject.getLong("ts");
            dataJsonObj.put("ts", ts);

            //将当前处理的维度数据以及这条数据对应的配置文件封装为Tuple2对象，传递到下游
            collector.collect(Tuple2.of(dataJsonObj,tableProcessDwd));
        }


    }


    @Override
    public void processBroadcastElement(TableProcessDwd tableProcessDwd, BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>>.Context context, Collector<Tuple2<JSONObject, TableProcessDwd>> collector) throws Exception {
        BroadcastState<String, TableProcessDwd> broadcastState = context.getBroadcastState(mapStateDescriptor);

        //获取对配置表进行操作的类型
        String op = tableProcessDwd.getOp();

        String sourceTable = tableProcessDwd.getSourceTable();
        String sourceType = tableProcessDwd.getSourceType();
        String key = getKey(sourceTable, sourceType);

        if ("d".equals(op)) {
            broadcastState.remove(key);
            configMap.remove(key);
        } else {
            broadcastState.put(key, tableProcessDwd);
            configMap.put(key, tableProcessDwd);
        }
    }

    private void deleteNotNeedColums(JSONObject dataJsonObj, String sinkColumns) {
        List<String> columnList = Arrays.asList(sinkColumns.split(","));

        Set<Map.Entry<String, Object>> entries = dataJsonObj.entrySet();

        entries.removeIf(entry -> !columnList.contains(entry.getKey()));
    }

    public String getKey(String sourceTable, String sourceType) {
        return sourceTable + ":" + sourceType;
    }
}
